package com.test.eventbus;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.SubscriberExceptionContext;
import com.google.common.eventbus.SubscriberExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 *
 *
 **/
public class SubscriberExceptionHandlerImpl implements SubscriberExceptionHandler {
    private Logger logger = LoggerFactory.getLogger(SubscriberExceptionHandlerImpl.class);

    ConcurrentHashMap<Integer,Model> concurrentHashMap = new ConcurrentHashMap();
    ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);

    public SubscriberExceptionHandlerImpl(){
        scheduledExecutorService.scheduleWithFixedDelay(new RemoveCache(), 1,10,TimeUnit.MINUTES);
        Model model = new Model("ok");
        model.setTime(System.currentTimeMillis()-60000);
        concurrentHashMap.put(new Integer(1),model);
    }

    @Override
    public void handleException(Throwable exception, SubscriberExceptionContext context) {
        logger.error("[EventBusError]:{}",exception);
        Object event = context.getEvent();
        if(event != null){
          Integer hashCode =  System.identityHashCode(event);
            Model model =  concurrentHashMap.get(hashCode);
            if ( model == null) {
                 model = new Model(event);
                concurrentHashMap.put(hashCode,model);
            }else{
                model.setRunTime(model.getRunTime()+1);
            }
            if(model.getRunTime() <5){
                EventBus eventBus = context.getEventBus();
                scheduledExecutorService.schedule(new Runnable() {
                    @Override
                    public void run() {
                        eventBus.post(event);
                    }
                },2, TimeUnit.SECONDS);
            } else {
                  concurrentHashMap.remove(hashCode);

            }
        }

    }

    public class Model{
        public Model( Object event){
            this.event = event;
            this.time = System.currentTimeMillis();
            this.runTime = 1;
        }
        long time = 0;
        long runTime = 0;
        Object event = null;

        public long getTime() {
            return time;
        }

        public void setTime(long time) {
            this.time = time;
        }

        public Object getEvent() {
            return event;
        }

        public void setEvent(Object event) {
            this.event = event;
        }

        public long getRunTime() {
            return runTime;
        }

        public void setRunTime(long runTime) {
            this.runTime = runTime;
        }
    }
    public class RemoveCache implements Runnable{
        @Override
        public void run() {
            long now = System.currentTimeMillis();
            Iterator<Integer> iterator = concurrentHashMap.keySet().iterator();
            while(iterator.hasNext()){
                Integer integer =  iterator.next();
               Model model =  concurrentHashMap.get(integer);
               if(model != null && (now - model.getTime()) > 60000){
                   concurrentHashMap.remove(integer);
               }
            }
        }
    }

}
